/*
 * Copyright 2020-2025 the original author or authors.
 * You cannot use this file unless authorized by the author.
 */
package org.ipig.messaging.kafka.producer;

import org.ipig.constants.ResultStatus;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.Serializable;

/**
 * GenericKafkaProducer
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id: GenericKafkaProducer.java 3 2020-01-21 09:36:33Z ts <comchnts@163.com> $$
 * @since 1.0
 */
@Slf4j
public class GenericKafkaProducer extends AbstractKafkaProducer implements KafkaProducerService, Serializable {
    /**
     * send
     *
     * @param message
     * @param key
     */
    @Override
    public ResultStatus send(String message, String key) {
        ResultStatus resultStatus = ResultStatus.SUCCESS;
        try {
            ProducerRecord<String, String> record = null;
            if (StringUtils.isBlank(key)) {
                // 如果客户端不指定Patition，也没有指定Key的话，使用自增长的数字取余数的方式实现指定的Partition。这样Kafka将平均的向Partition中生产数据
                record = new ProducerRecord<String, String>(getConf().getTopicName(), message);
            } else {
                // 如果客户端不指定Patition，也没有指定Key的话，使用自增长的数字取余数的方式实现指定的Partition。这样Kafka将平均的向Partition中生产数据
                record = new ProducerRecord<String, String>(getConf().getTopicName(), key, message);
            }
//            Future<RecordMetadata> result=getProducer().send(record);
            getProducer().send(record);
            if (getBeginTimeMills() <= 0) {
                setBeginTimeMills(System.currentTimeMillis());
            }
            getTotal().increment();
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage(), e);
            resultStatus = ResultStatus.FAILURE;
        }
        return resultStatus;
    }

}
